package com.shujia.mr.output;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

// 由于需要写出的数据是从Map端传入的 所以RecordWriter的泛型为Map端输出的数据类型
public class MyRecordWriter extends RecordWriter<Text,Text> {

    FSDataOutputStream fsDataOutputStreamMore450;
    FSDataOutputStream fsDataOutputStreamLess450;
    FileSystem fileSystem;

    public MyRecordWriter() {
    }


    /**
     * 如果要写出数据，那么必须要创建输出流，创建输出流可以通过FileSystem来进行
     *      所以第一步可以先创建FileSystem对象，在什么地方创建，
     *      在源码中写出数据是需要通过调用MyRecordWriter对象的write函数，
     *      那么可以在MyRecordWriter构造方法中进行处理
     *      而创建FileSystem对象需要获取Configration对象才能获取
     */

    public MyRecordWriter(Configuration configuration) throws IOException {
        fileSystem = FileSystem.get(configuration);
        fsDataOutputStreamMore450 = fileSystem.create(new Path("output/output/more450.txt"));
        fsDataOutputStreamLess450 = fileSystem.create(new Path("output/output/less450.txt"));

    }


    @Override
    public void write(Text key, Text value) throws IOException, InterruptedException {
        if (Integer.parseInt(value.toString()) > 450){
            fsDataOutputStreamMore450.writeUTF(key.toString()+","+value.toString()+"\n");
        }else{
            fsDataOutputStreamLess450.writeUTF(key.toString()+","+value.toString()+"\n");
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        fsDataOutputStreamMore450.close();
        fsDataOutputStreamLess450.close();
        fileSystem.close();
    }
}
